

namespace Stee.CAP8.MessageBus.Kafka.Configurations
{
    using System;
    using System.Collections.Generic;
    using Confluent.Kafka;
    using Stee.CAP8.MessageBus.Interface.Utility;

    /// <summary>
    /// Configuration for Kafka producer
    /// See below reference for detailed documentation on indivdual configurations
    /// https://docs.confluent.io/current/installation/configuration/producer-configs.html
    /// </summary>
    public class PublisherConfiguration
    {
        #region Properties

        /// <summary>
        /// Gets the initial bootstrap servers' addresses as a CSV list of broker host or host:port.
        /// Mandatory configuration.
        /// </summary>
        public string BootstrapServers { get; set; } = string.Empty;

        /// <summary>
        /// Gets or sets the number of acknowledgements the leader broker must receive
        /// from ISR brokers before responding to the request: Zero=Broker does not send
        /// any response/ack to client, One=The leader will write the record to its local
        /// log but will respond without awaiting full acknowledgement from all followers.
        /// All=Broker will block until message is committed by all in sync replicas (ISRs).
        /// If there are less than min.insync.replicas (broker configuration) in the ISR
        /// set the produce request will fail.
        /// Values: All, None, Leader
        /// </summary>
        public Acks Acks { get; set; } = Acks.All;

        /// <summary>
        /// Gets or sets the client identifier
        /// </summary>
        public string ClientID { get; set; } = null;

        /// <summary>
        /// Gets or sets the flag to ensure that messages are successfully produced
        /// exactly once and in the original produce order. The following configuration properties
        /// are adjusted automatically (if not modified by the user) when idempotence is
        /// enabled: 'max.in.flight.requests.per.connection=5' (must be less than or equal
        /// to 5), 'retries=INT32_MAX' (must be greater than 0), 'acks=all', 'queuing.strategy=fifo'.
        /// Producer instantation will fail if user-supplied configuration is incompatible.
        /// If you have EnableIdempotence set to true, Acks must be all.
        /// default: false importance: high
        /// </summary>
        public bool EnableIdempotence { get; set; } = true;

        /// <summary>
        /// Gets or sets the delay in milliseconds to wait for messages in the producer queue to accumulate
        /// before constructing message batches (MessageSets) to transmit to brokers. A higher
        /// value allows larger and more effective (less overhead, improved compression)
        /// batches of messages to accumulate at the expense of increased message delivery
        /// latency. A good general purpose setting is 5 (the default is 0.5).
        /// default: 0.5 importance: high
        /// </summary>
        public double LingerMs { get; set; } = 0.5;

        /// <summary>
        /// Gets or sets the number of times to retry sending a failing Message.
        /// **Note:** retrying may cause reordering unless 'enable.idempotence' is set to true.
        /// default: 2 importance: high
        /// </summary>
        public int MessageSendMaxRetries { get; set; } = 2;

        /// <summary>
        /// Gets or sets the local message timeout.
        /// This value is only enforced locally and limits the time
        /// a produced message waits for successful delivery. A time of 0 is infinite. This
        /// is the maximum time librdkafka may use to deliver a message (including retries).
        /// Delivery error occurs when either the retry count or the message timeout are
        /// exceeded. The message timeout is automatically adjusted to 'transaction.timeout.ms'
        /// if 'transactional.id' is configured.
        /// default: 300000 importance: high
        /// </summary>
        public int MessageTimeoutMs { get; set; } = 300000;

        /// <summary>
        /// Gets or sets the maximum total message size sum allowed on the producer queue. This queue is shared
        /// by all topics and partitions. This property has higher priority than queue.buffering.max.messages.
        /// default: 1048576 importance: high
        /// </summary>
        public int QueueBufferingMaxKbytes { get; set; } = 1048576;

        /// <summary>
        /// Gets or sets  the maximum number of messages allowed on the producer queue. This queue is shared
        /// by all topics and partitions.
        /// default: 100000 importance: high
        /// </summary>
        public int QueueBufferingMaxMessages { get; set; } = 100000;

        /// <summary>
        /// Gets or sets the maximum amount of time in milliseconds that the transaction coordinator will wait
        /// for a transaction status update from the producer before proactively aborting the ongoing transaction.
        /// If this value is larger than the 'transaction.max.timeout.ms' setting in the broker, the init_transactions()
        /// call will fail with ERR_INVALID_TRANSACTION_TIMEOUT. The transaction timeout automatically adjusts
        /// 'message.timeout.ms' and 'socket.timeout.ms', unless explicitly configured in which case they must not exceed
        /// the transaction timeout ('socket.timeout.ms' must be at least 100ms lower than 'transaction.timeout.ms').
        /// This is also the default timeout value if no timeout (-1) is supplied to the transactional API methods.
        /// default: 60000 importance: medium
        /// </summary>
        public int TransactionTimeoutMs { get; set; } = 60000;

        /// <summary>
        /// Gets the validity of subscriber configuration
        /// </summary>
        public bool IsValid
        {
            get { return this.Validate(); }
        }

        #endregion Properties


        #region Constructor & Initializers

        /// <summary>
        /// Initializes a new instance of the PublisherConfiguration class
        /// </summary>
        /// <param name="bootstrapServers">Bootstrap servers' addresses as a CSV list</param>
        /// <param name="acks"></param>
        public PublisherConfiguration(string bootstrapServers, Acks acks)
        {
            bootstrapServers = bootstrapServers?.Trim();

            if (string.IsNullOrEmpty(bootstrapServers))
            {
                throw new ArgumentNullException(nameof(bootstrapServers));
            }

            this.BootstrapServers = bootstrapServers;
            this.Acks = acks;
        }

        /// <summary>
        /// Initializes a new instance of the SubscriberConfiguration class.
        /// Make certain that all cases of dictionary keys matches that of the respective class properties.
        /// </summary>
        /// <param name="configurations">Configuration mappings to property name. Eg: "BootstrapServers" => "127.0.0.1:9092"</param>
        public PublisherConfiguration(Dictionary<string, string> configurations)
        {
            if (configurations == null)
            {
                throw new ArgumentNullException($"Configuration is null: {nameof(configurations)}");
            }
            else if (!configurations.ContainsKey(nameof(this.BootstrapServers)))
            {
                throw new ArgumentNullException($"Mandatory configuration is null: {nameof(this.BootstrapServers)}");
            }

            //// Check mandatory configurations
            string bootstrapServers = configurations[nameof(this.BootstrapServers)];

            if (string.IsNullOrEmpty(bootstrapServers))
            {
                throw new ArgumentNullException($"Mandatory configuration is empty: {nameof(this.BootstrapServers)}");
            }

            this.InitializeProperties(configurations);
        }

        /// <summary>
        /// Initializes a new instance of the PublishConfiguration class
        /// </summary>
        /// <param name="currentConfig">Current publisher configuration</param>
        private PublisherConfiguration(PublisherConfiguration currentConfig)
        {
            this.Acks = currentConfig.Acks;
            this.BootstrapServers = currentConfig.BootstrapServers;
            this.ClientID = currentConfig.ClientID;
            this.EnableIdempotence = currentConfig.EnableIdempotence;
            this.LingerMs = currentConfig.LingerMs;
            this.MessageSendMaxRetries = currentConfig.MessageSendMaxRetries;
            this.MessageTimeoutMs = currentConfig.MessageTimeoutMs;
            this.QueueBufferingMaxKbytes = currentConfig.QueueBufferingMaxKbytes;
            this.QueueBufferingMaxMessages = currentConfig.QueueBufferingMaxMessages;
            this.TransactionTimeoutMs = currentConfig.TransactionTimeoutMs;
        }

        /// <summary>
        /// Initializes the configuration mappings into properties
        /// </summary>
        /// <param name="configurations">Configuration mappings to property name. Eg: "BootstrapServers" => "127.0.0.1:9092"</param>
        private void InitializeProperties(Dictionary<string, string> configurations)
        {
            foreach (KeyValuePair<string, string> item in configurations)
            {
                switch (item.Key)
                {
                    case nameof(this.BootstrapServers):
                        this.BootstrapServers = item.Value?.Trim();
                        break;

                    case nameof(this.Acks):
                        this.Acks = ValueConverter.ConvertToEnum<Acks>(item.Value, this.Acks);
                        break;

                    case nameof(this.ClientID):
                        this.ClientID = item.Value?.Trim();
                        break;

                    case nameof(this.EnableIdempotence):
                        this.EnableIdempotence = ValueConverter.ConvertToBoolean(item.Value, this.EnableIdempotence);
                        break;

                    case nameof(this.LingerMs):
                        this.LingerMs = ValueConverter.ConvertToDouble(item.Value, this.LingerMs);
                        break;

                    case nameof(this.MessageSendMaxRetries):
                        this.MessageSendMaxRetries = ValueConverter.ConvertToInteger(item.Value, this.MessageSendMaxRetries);
                        break;

                    case nameof(this.MessageTimeoutMs):
                        this.MessageTimeoutMs = ValueConverter.ConvertToInteger(item.Value, this.MessageTimeoutMs);
                        break;

                    case nameof(this.QueueBufferingMaxKbytes):
                        this.QueueBufferingMaxKbytes = ValueConverter.ConvertToInteger(item.Value, this.QueueBufferingMaxKbytes);
                        break;

                    case nameof(this.QueueBufferingMaxMessages):
                        this.QueueBufferingMaxMessages = ValueConverter.ConvertToInteger(item.Value, this.QueueBufferingMaxMessages);
                        break;

                    case nameof(this.TransactionTimeoutMs):
                        this.TransactionTimeoutMs = ValueConverter.ConvertToInteger(item.Value, this.TransactionTimeoutMs);
                        break;

                    default:
                        //// TODO: log invalid property
                        break;
                }
            }
        }

        #endregion Constructor & Initializers


        #region Public Methods

        /// <summary>
        /// Clone a new instance of PublisherConfiguration class
        /// </summary>
        /// <returns>New PublisherConfiguration instance</returns>
        public PublisherConfiguration Clone()
        {
            return new PublisherConfiguration(this);
        }

        #endregion Public Methods


        #region Private Methods

        /// <summary>
        /// Validate publisher configurations
        /// </summary>
        /// <returns>Return TRUE if configuration is valid else FALSE</returns>
        private bool Validate()
        {
            bool isValid = true;

            if (string.IsNullOrEmpty(this.BootstrapServers) ||
                this.LingerMs < 0.1 ||
                this.MessageSendMaxRetries < 1 ||
                this.MessageTimeoutMs < 1000 ||
                this.QueueBufferingMaxKbytes < 1000 ||
                this.QueueBufferingMaxMessages < 1 ||
                this.TransactionTimeoutMs < 1000)
            {
                isValid = false;
            }

            return isValid;
        }

        #endregion Private Methods
    }
}
